# ******************************************************************#
##author: v_sunao3
##create time: 2023-10-25 10:37:03
# ******************************************************************#


# !/usr/bin/env python3
import sys
import time
from datetime import datetime
from pyhive import hive
import pymysql

# 数据时间通过变量传入
yesterday = '${yyyy-MM-dd-1d}'
print("********************业务日期为：" + str(yesterday))
# 获取当前时间戳
current_timestamp = int(time.time())
# 获取当前时间
current_datetime = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print("********************当前任务运行时间：" + str(current_datetime))


# 获取hive连接
def conn_to_hive(hive_database,
                 hive_host='172.25.1.64',
                 hive_port=7001,
                 hive_user='v_sunao3'
                 ):
    return hive.Connection(
        host=hive_host,
        port=hive_port,
        username=hive_user,
        database=hive_database
    )


# 连接mysql数据库
def conn_to_mysql(mysql_database,
                  mysql_host='172.25.1.58',
                  mysql_port=3306,
                  mysql_user='oceanus_user',
                  mysql_password='sqYcvrHc5gpEBH2a',
                  mysql_charset='utf8mb4',
                  ):
    return pymysql.connect(
        host=mysql_host,
        port=mysql_port,
        user=mysql_user,
        password=mysql_password,
        database=mysql_database,
        charset=mysql_charset
    )


# 连接starrocks数据库
def conn_to_starrocks(starrocks_host='172.25.1.67',
                      starrocks_port=9030,
                      starrocks_user='test',
                      starrocks_password='srLcHLyUjUrvFf6R',
                      starrocks_database='test',
                      starrocks_charset='utf8mb4'
                      ):
    return pymysql.connect(
        host=starrocks_host,
        port=starrocks_port,
        user=starrocks_user,
        password=starrocks_password,
        database=starrocks_database,
        charset=starrocks_charset
    )


# 连接Hive，计算表数据量,传入状态码
def compute_hive_count(hive_table, cursor_hive, stat_date='${yyyyMMdd-1d}'):
    result = None
    print("********************查询hive表的分区日期为：" + str(stat_date))
    try:
        # 获取Hive表数据量
        sql = f"SELECT COUNT(*) FROM {hive_table} where stat_date='{stat_date}'"
        # print(f"********************执行计算的sql：{sql}")
        cursor_hive.execute(sql)
        result = cursor_hive.fetchone()[0]

    except Exception as e:
        print("=" * 100)
        print(f"Failed to connect to Hive: {str(e)}")
        print("=" * 100)
    finally:
        cursor_hive.close()

    return result


def compute_mysql_count(mysql_table, cursor_mysql):
    result = None
    try:
        # 获取mysql表的数据量
        cursor_mysql.execute(f"SELECT COUNT(*) FROM {mysql_table}")
        result = cursor_mysql.fetchone()[0]

    except Exception as e:
        print("=" * 100)
        print(f"Failed to connect to Mysql: {str(e)}")
        print("=" * 100)
    finally:
        cursor_mysql.close()

    return result


def check_hive_mysql_data_count(hive_table, mysql_table, conn_mysql, conn_hive):
    # 计算hive表数据量
    result = None
    cursor_mysql = conn_mysql.cursor()
    cursor_hive = conn_hive.cursor()
    hive_data_count = compute_hive_count(cursor_hive=cursor_hive, hive_table=hive_table)
    mysql_data_count = compute_mysql_count(cursor_mysql=cursor_mysql, mysql_table=mysql_table)
    if hive_data_count is None:
        result = [f"{hive_table} 还未同步", 0,
                  [mysql_table, hive_table, '${yyyyMMdd-1d}', 0, hive_data_count, mysql_data_count, None,"不通过",
                   current_datetime]]

    if hive_data_count == mysql_data_count:

        result = [f"********************{hive_table}:{hive_data_count}和{mysql_table}:{mysql_data_count} 数据量一致", 1,
                  [mysql_table, hive_table, '${yyyyMMdd-1d}', 1, hive_data_count, mysql_data_count, '是',"通过",
                   current_datetime]]
    else:
        result = [
            f"********************{hive_table}:{hive_data_count}和{mysql_table}:{mysql_data_count} 数据量不一致,请检查原因，并重新同步数据",
            0,
            [mysql_table, hive_table, '${yyyyMMdd-1d}', 1, hive_data_count, mysql_data_count, '不',"不通过", current_datetime]]

    return result


def get_mysql_tables(cursor_mysql):
    try:
        # 获取mysql当前库的所有表
        cursor_mysql.execute("show tables")
        return cursor_mysql.fetchall()

    except Exception as e:
        print(f"Failed to connect to Mysql: {str(e)}")
        return None
    finally:
        cursor_mysql.close()

if __name__ == "__main__":

    hive_database = 'ods_sg'
    mysql_database = 'rent'

    # 链接数据库并创建游标
    conn_hive = conn_to_hive(hive_database=hive_database)
    conn_mysql = conn_to_mysql(mysql_database=mysql_database)
    conn_sr = conn_to_starrocks()

    mysql_tables = get_mysql_tables(conn_mysql.cursor())

    table_count = len(list(mysql_tables))
    complete_cnt = 0
    # print(f"********************总计：{table_count}张表进行稽核")
    if mysql_tables is not None:
        for mysql_table in list(mysql_tables):
            hive_table = f'ods_sg_zl_{mysql_table[0]}_df'
            result = check_hive_mysql_data_count(hive_table=hive_table,
                                                 mysql_table=mysql_table[0],
                                                 conn_mysql=conn_mysql,
                                                 conn_hive=conn_hive)

            # 插入数据
            cursor_sr = conn_sr.cursor()
            try:

                mysql_table_name,hive_table_name,event_date,mysql2hive_status,hive_data_num,mysql_data_num,is_consistent,is_pass,script_run_time = result[2]

                sql = f"""INSERT OVERWRITE python_mysql_hive_ods_data_cnt_df 
                    (mysql_table_name,hive_table_name,event_date,mysql2hive_status,hive_data_num,mysql_data_num,is_consistent,script_run_time,is_pass) values
                    ("{mysql_table_name}","{hive_table_name}","{event_date}",{mysql2hive_status},{hive_data_num},{mysql_data_num},"{is_consistent}","{script_run_time}","{is_pass}")
                    """
                # print("********************", sql)
                cursor_sr.execute(sql)
                conn_sr.commit()
                # print('********************Data inserted successfully.')

            except Exception as e:
                conn_sr.rollback()
                print('********************Failed to insert data:', str(e))
            finally:

                # 关闭游标和连接
                cursor_sr.close()

            complete_cnt += int(result[1])
            # print(f"********************总计：{table_count}张表，已完成：{complete_cnt}个表的数据量的检测")

    conn_mysql.close()
    conn_hive.close()
    conn_sr.close()
